package com.hezhiqin.kafkademo;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @program: homeWork
 * @description:
 * @author: hezhiqin
 * @create: 2020-02-26 13:20
 */
public class KafkaTest extends Thread {
    KafkaProducer<Integer,String> kafkaProducer ;
    String topic;

    public KafkaTest(String topic) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.223.128:9092,192.168.223.129:9092,192.168.223.130:9092");
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "hezhiqin-kafkaProducer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        kafkaProducer = new KafkaProducer<Integer, String>(properties);
        this.topic = topic;
    }


    @Override
    public void run() {
        int num = 0;
        String msg = "hello kafka";
        while(num<20){
            try {
                RecordMetadata recordMetadata =  kafkaProducer.send(new ProducerRecord<Integer,String>(topic,msg)).get();
                System.out.println(recordMetadata.offset()+"->"+recordMetadata.partition()+"_>"+recordMetadata.topic());
              
                TimeUnit.SECONDS.sleep(2L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }

    }

    public static void main(String[] args) {
        new KafkaTest("kafkaTest").start();
    }
}
